package base

import (
	"errors"
	"sync/atomic"
)

type RingBufferAtomic struct {
	buf    []interface{}
	size   int64
	rPos   int64
	wPos   int64
	closed int32
}

func NewRingBufferAtomic(size int) *RingBufferAtomic {
	return &RingBufferAtomic{
		buf:  make([]interface{}, size),
		size: int64(size),
	}
}

func (rb *RingBufferAtomic) Put(value interface{}) error {
	for {
		if atomic.LoadInt32(&rb.closed) == 1 {
			return errors.New("ringbuffer closed")
		}

		wPos := atomic.LoadInt64(&rb.wPos)
		rPos := atomic.LoadInt64(&rb.rPos)

		if (wPos+1)%rb.size == rPos {
			// buffer is full
			continue
		}

		if atomic.CompareAndSwapInt64(&rb.wPos, wPos, (wPos+1)%rb.size) {
			rb.buf[wPos] = value
			return nil
		}
	}
}

func (rb *RingBufferAtomic) Get() (interface{}, error) {
	for {
		if atomic.LoadInt32(&rb.closed) == 1 && rb.isEmpty() {
			return nil, errors.New("ringbuffer closed and empty")
		}

		rPos := atomic.LoadInt64(&rb.rPos)
		wPos := atomic.LoadInt64(&rb.wPos)

		if rPos == wPos {
			// buffer is empty
			if atomic.LoadInt32(&rb.closed) == 1 {
				return nil, errors.New("ringbuffer closed and empty")
			}
			continue
		}

		if atomic.CompareAndSwapInt64(&rb.rPos, rPos, (rPos+1)%rb.size) {
			val := rb.buf[rPos]
			rb.buf[rPos] = nil // 避免内存泄漏
			return val, nil
		}
	}
}

func (rb *RingBufferAtomic) Close() {
	atomic.StoreInt32(&rb.closed, 1)
}

func (rb *RingBufferAtomic) isEmpty() bool {
	return atomic.LoadInt64(&rb.rPos) == atomic.LoadInt64(&rb.wPos)
}

func (rb *RingBufferAtomic) Len() int64 {
	wPos := atomic.LoadInt64(&rb.wPos)
	rPos := atomic.LoadInt64(&rb.rPos)

	if wPos >= rPos {
		return wPos - rPos
	}
	return rb.size - rPos + wPos
}
